// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#include "vec/data_types/data_type_fixed_length_object.h"

#include <glog/logging.h>
#include <string.h>

#include <ostream>

#include "agent/be_exec_version_manager.h"
#include "common/cast_set.h"
#include "vec/columns/column.h"
#include "vec/common/assert_cast.h"
#include "vec/core/types.h"

namespace doris::vectorized {
#include "common/compile_check_begin.h"

char* DataTypeFixedLengthObject::serialize(const IColumn& column, char* buf,
                                           int be_exec_version) const {
    // const flag
    bool is_const_column = is_column_const(column);
    unaligned_store<bool>(buf, is_const_column);
    buf += sizeof(bool);

    // row num
    const auto row_num = column.size();
    unaligned_store<size_t>(buf, row_num);
    buf += sizeof(size_t);
    auto real_need_copy_num = is_const_column ? 1 : row_num;

    const IColumn* data_column = &column;
    if (is_const_column) {
        const auto& const_column = assert_cast<const ColumnConst&>(column);
        data_column = &(const_column.get_data_column());
    }
    const auto& src_col = assert_cast<const ColumnType&>(*data_column);
    DCHECK(src_col.item_size() > 0)
            << "[serialize]item size of DataTypeFixedLengthObject should be greater than 0";

    // item size
    unaligned_store<size_t>(buf, src_col.item_size());
    buf += sizeof(size_t);
    // column data
    const auto* origin_data = src_col.get_data().data();
    memcpy(buf, origin_data, real_need_copy_num * src_col.item_size());
    buf += real_need_copy_num * src_col.item_size();

    return buf;
}

const char* DataTypeFixedLengthObject::deserialize(const char* buf, MutableColumnPtr* column,
                                                   int be_exec_version) const {
    //const flag
    bool is_const_column = unaligned_load<bool>(buf);
    buf += sizeof(bool);
    //row num
    size_t row_num = unaligned_load<size_t>(buf);
    buf += sizeof(size_t);
    //item size
    size_t item_size = unaligned_load<size_t>(buf);
    buf += sizeof(size_t);

    DCHECK(item_size > 0)
            << "[deserialize]item size of DataTypeFixedLengthObject should be greater than 0";

    auto& dst_col = static_cast<ColumnType&>(*(column->get()));
    dst_col.set_item_size(item_size);
    // column data
    auto real_copy_num = is_const_column ? 1 : row_num;
    dst_col.resize(real_copy_num);
    memcpy(dst_col.get_data().data(), buf, real_copy_num * item_size);
    buf += real_copy_num * item_size;
    if (is_const_column) {
        auto const_column = ColumnConst::create((*column)->get_ptr(), row_num);
        *column = const_column->get_ptr();
    }
    return buf;
}

// binary: const flag | row num | item size| data
// data  : item data1 | item data2...
int64_t DataTypeFixedLengthObject::get_uncompressed_serialized_bytes(const IColumn& column,
                                                                     int be_exec_version) const {
    auto size = sizeof(bool) + sizeof(size_t) + sizeof(size_t);
    const IColumn* data_column = &column;
    if (is_column_const(column)) {
        const auto& const_column = assert_cast<const ColumnConst&>(column);
        data_column = &(const_column.get_data_column());
    }
    const auto& src_col = assert_cast<const ColumnType&>(*data_column);
    return static_cast<const ColumnType&>(src_col).byte_size() + size;
}

MutableColumnPtr DataTypeFixedLengthObject::create_column() const {
    return ColumnType::create(0);
}

Status DataTypeFixedLengthObject::check_column(const IColumn& column) const {
    return check_column_non_nested_type<ColumnType>(column);
}

} // namespace doris::vectorized